package com.pw.study.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import scala.compat.java8.JFunction0$mcD$sp;

/**
 * @Author: linux_future
 * @since: 2022/3/12
 **/
public class $17SqlTopN {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        tEnv.executeSql(
                "create table user_behavior(" +
                        "   user_id bigint, " +
                        "   item_id bigint, " +
                        "   category_id int, " +
                        "   behavior string, " +
                        "   ts bigint, " +
                        "   event_time as to_timestamp(from_unixtime(ts, 'yyyy-MM-dd HH:mm:ss')), " +
                        "   watermark for event_time as  event_time - interval '5' second " +
                        ")with(" +
                        "   'connector'='filesystem', " +
                        "   'path'='data/file/UserBehavior.csv', " +
                        "   'format'='csv')"
        );
        // 1. 计算每每个窗口内每个商品的点击量
        //mc1(tEnv);
        Table t1 = mc11(tEnv);
        tEnv.createTemporaryView("t1", t1);

        // 2. 按照窗口开窗, 对商品点击量进行排名
        Table t2 = mc2(tEnv);
        //top3
        Table t3=mc3(tEnv);
        //4.写入mysql数据
        tEnv.executeSql("CREATE TABLE `hot_item` (" +
                "  `w_end` timestamp," +
                "  `item_id` bigint," +
                "  `item_count` bigint," +
                "  `rk` bigint," +
                "  PRIMARY KEY (`w_end`,`rk`)not enforced" +
                ")with(" +
                "   'connector' = 'jdbc', " +
                "   'url' = 'jdbc:mysql://hadoop162:3306/flink_sql?useSSL=false'," +
                "   'table-name' = 'hot_item', " +
                "   'username' = 'root', " +
                "   'password' = 'aaaaaa'" +
                ")");

        // 5.2 把结果写入到动态表, 则自动会写入到mysql
        t3.executeInsert("hot_item");

    }

    private static Table mc3(StreamTableEnvironment tEnv) {
        System.out.println("top3......");
        //重点：必须加入where条件 不然会报错The window can only be ordered in ASCENDING mode.
        //tEnv.sqlQuery("select * from t2 ").execute().print();
        Table table = tEnv.sqlQuery("select w_end,item_id,ct item_count,rk from t2 where rk <3");
        table.execute().print();
        return table;
    }

    private static Table mc2(StreamTableEnvironment tEnv) {
        System.out.println("=====2. 按照窗口开窗, 对商品点击量进行排名");
        //重点：flink只有row_number
        Table table = tEnv.sqlQuery("select * ," +
                " row_number() over(partition by w_end order by ct desc ) rk" +
                " from t1");

        tEnv.createTemporaryView("t2", table);
        //重点：必须加入where条件 不然会报错The window can only be ordered in ASCENDING mode.
        //tEnv.sqlQuery("select * from t2 ").execute().print();
        tEnv.sqlQuery("select item_id,w_end,ct,rk from t2 where rk <3").execute().print();

        return table;
    }

    private static Table mc11(StreamTableEnvironment tEnv) {
        System.out.println("===== 2. 计算每每个窗口内每个商品的点击量优化sql======");
        Table table = tEnv.sqlQuery("select " +
                " item_id," +
                " window_start startTime," +
                " window_end w_end," +
                " count(*) ct" +
                " from table(" +
                " hop(table user_behavior,descriptor(event_time),interval '1' hour ,interval '2' hour)" +
                " ) where  behavior='pv'" +
                " group by item_id,window_start,window_end ");

        // table.execute().print();
        return table;
    }

    private static void mc1(StreamTableEnvironment tEnv) {
        System.out.println("===== 1. 计算每每个窗口内每个商品的点击量======");
        tEnv.sqlQuery("select " +
                " item_id," +
                " hop_start(event_time,interval '1' hour ,interval '2' hour) sta," +
                " hop_end(event_time,interval '1' hour ,interval '2' hour) ed," +
                " count(*) ct" +
                " from user_behavior where behavior='pv' group by item_id, " +
                " hop(event_time, interval '1' hour ,interval '2' hour) ").execute().print();
    }

}
